package org.fastsyncer.parser.mysql;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.List;
import java.util.Map;
import java.util.Queue;

import org.fastsyncer.common.constant.ConnectorConstant;
import org.fastsyncer.common.entity.DatabaseConfig;
import org.fastsyncer.common.entity.Mapping;
import org.fastsyncer.common.entity.MappingTask;
import org.fastsyncer.connector.util.DatabaseUtil;
import org.fastsyncer.parser.framework.AbstractDatabaseParser;
import org.json.JSONException;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jdbc.core.BatchPreparedStatementSetter;
import org.springframework.jdbc.core.JdbcTemplate;

// 声明一个final，使别人不能随便new对象
// 它只被调用一次，实例存在之后就不再调用了
public final class MysqlParser extends AbstractDatabaseParser {

    // 实现单例模式的方法，通过方法来实现实例化
    // 声明一个私有的静态属性，使只被实例化一次
    private static volatile MysqlParser instance = null;

    private MysqlParser() {
    }

    // 声明一个公共的静态方法，外界通过调用这个方法来得到唯一的实例
    public static MysqlParser getInstance() {
        if (null == instance) {
            synchronized (MysqlParser.class) {
                if (null == instance) {
                    instance = new MysqlParser();
                }
            }
        }
        return instance;
    }

    private static Logger logger = LoggerFactory.getLogger(MysqlParser.class);

    @Override
    public List<Map<String, Object>> batchQuery(MappingTask mappingTask, Integer startPos, Integer endPos) throws Exception {
        Mapping sourceMapping = mappingTask.getSourceMapping();
        // 1.获取源所有数据,例如 [{ID=123, NAME=张三11}, {ID=321, NAME=李四},...]
        // 数据源实例
        DatabaseConfig sourceConfig = (DatabaseConfig) sourceMapping.getConfig();
        JdbcTemplate sourceJdbc = super.getJdbcTemplate(sourceConfig);
        String sql = sourceMapping.getExecuteCommond().get(ConnectorConstant.OPERTION_QUERY);
        Object[] args = new Object[] { startPos, endPos };
        List<Map<String, Object>> rows = sourceJdbc.queryForList(sql, args);

        // 释放连接
        DatabaseUtil.close(sourceJdbc);
        return rows;
    }

    @Override
    public Integer[] batchInsert(MappingTask mappingTask, List<Map<String, Object>> rows, Integer[] pos, Queue<JSONObject> reasonQuene) throws Exception {
        Mapping targetMapping = mappingTask.getTargetMapping();

        // 去掉目标重复的映射字段
        List<Map<String, Object>> targetFileds = this.filterFields(targetMapping.getFileds());

        // 批量条数
        int size = rows.size();
        BatchPreparedStatementSetter batchSetter = new BatchPreparedStatementSetter() {
            @Override
            public void setValues(PreparedStatement ps, int i) throws SQLException {
                // 2.获取源同步字段sourcePro:[{name=ID, type=4}, {name=NAME, type=12}]
                // 获取每行中的所有字段值，例如 {ID=123, NAME=张三11}
                batchRowsSetter(ps, targetFileds, rows.get(i));
            }

            @Override
            public int getBatchSize() {
                return size;
            }
        };

        // 总数和成功数
        int success = 0;
        try {
            // 目标源实例
            DatabaseConfig targetConfig = (DatabaseConfig) targetMapping.getConfig();
            JdbcTemplate targetJdbc = super.getJdbcTemplate(targetConfig);
            // 执行批量同步
            String sql = targetMapping.getExecuteCommond().get(ConnectorConstant.OPERTION_INSERT);
            int[] batchUpdate = targetJdbc.batchUpdate(sql, batchSetter);
            for (int i = 0; i < size; i++) {
                success += (batchUpdate[i] == 1) ? 1 : 0;
            }
            // 释放连接
            super.close(targetJdbc);
        } catch (Exception e) {
            try {
                // 记录错误信息
                JSONObject err = new JSONObject();
                if (pos != null && pos.length == 2) {
                    err.put("startPos", pos[0]);
                    err.put("endPos", pos[1]);
                }
                err.put("message", e.getMessage());
                reasonQuene.offer(err);
            } catch (JSONException e1) {
                logger.error("throw an error message: Converts a JSONObject exception!");
            }
        }
        return new Integer[] { size, success };
    }

}
